package com.hzccz.kafka;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.time.Instant;
import java.util.Properties;

public class KafkaFileProducer {
    private final KafkaProducer<String, String> producer;
    private final String topic;

    public KafkaFileProducer(String bootstrapServers, String topic) {
        this.topic = topic;

        Properties properties = new Properties();
        
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        properties.setProperty(ProducerConfig.ACKS_CONFIG, ConfigUtils.getProperty("producer.acks", "all"));
        properties.setProperty(ProducerConfig.RETRIES_CONFIG, ConfigUtils.getProperty("producer.retries", "3"));
        properties.setProperty(ProducerConfig.LINGER_MS_CONFIG, ConfigUtils.getProperty("producer.linger.ms", "1"));

        String batchSize = ConfigUtils.getProperty("producer.batch.size");
        if (batchSize != null) {
            properties.setProperty(ProducerConfig.BATCH_SIZE_CONFIG, batchSize);
        }

        String bufferMemory = ConfigUtils.getProperty("producer.buffer.memory");
        if (bufferMemory != null) {
            properties.setProperty(ProducerConfig.BUFFER_MEMORY_CONFIG, bufferMemory);
        }

        this.producer = new KafkaProducer<>(properties);
    }

    public void sendFromFile(String filePath, long delayMs) throws IOException, InterruptedException {
        System.out.println("开始从文件发送数据: " + filePath);

        try (BufferedReader reader = new BufferedReader(new FileReader(filePath))) {
            String line;
            int count = 0;
            long startTime = Instant.now().toEpochMilli();

            while ((line = reader.readLine()) != null) {
                
                ProducerRecord<String, String> record = new ProducerRecord<>(topic, null, line);
                producer.send(record, (metadata, exception) -> {
                    if (exception != null) {
                        System.err.println("发送消息失败: " + exception.getMessage());
                    }
                });

                count++;
                if (count % 100 == 0) {
                    System.out.println("已发送 " + count + " 条消息");
                }

                if (delayMs > 0) {
                    Thread.sleep(delayMs);
                }
            }

            long endTime = Instant.now().toEpochMilli();
            System.out.println("发送完成，共发送 " + count + " 条消息，耗时 " + (endTime - startTime) + "ms");
        }
    }

    public void close() {
        if (producer != null) {
            producer.flush();
            producer.close();
        }
    }
}
